package com.huatai.datacenter.config;

import com.huatai.common.util.ObjectUtil;
import com.huatai.datacenter.entity.ClusterEntity;
import com.huatai.datacenter.entity.RabbitMQEntity;
import com.huatai.datacenter.entity.TopicTaskEntity;
import com.huatai.datacenter.job.InterfaceListener;
import com.huatai.datacenter.job.KafkaConsumerListener;
import com.huatai.datacenter.job.KafkaListenerJob;
import com.huatai.datacenter.service.*;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.context.WebApplicationContext;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author Lion
 * @date 2023/6/13  9:17
 */
@Configuration
@Slf4j
public class KafkaConfig {

	@Autowired
	private KafkaConsumerListener kafkaConsumerListener;

	@Autowired
	private KafkaConsumerService kafkaConsumerService;

	@Autowired
	private TopicTaskService topicTaskService;

	@Autowired
	private ClusterService clusterService;

	@Autowired
	private AllURLService allURLService;

	@Autowired
	private WebApplicationContext applicationContext;

	@Autowired
	private RabbitMQService rabbitMQService;

	List<TopicTaskEntity> dataList = new ArrayList<>();


	@Bean
	public void loadKafkaConfig() {



		List<TopicTaskEntity> listEntity = topicTaskService.list();
		//启动程序即对已创建的主题任务进行实时监听
		dataList = listEntity;
		compareList(listEntity);
	}

	/**
	 * @author Lion
	 * @description 比较两个集合是否一致
	 * @param list
	 * @return void
	 */
	public void compareList(List<TopicTaskEntity> list){

		//比较两个list是否相同，就按源数据订阅，不同则按新的集合订阅
		if(list.containsAll(dataList)){
			startListener(dataList);
		}else {
			startListener(list);
		}
	}



	/**
	 * @author Lion
	 * @description 启动实时监听
	 * @param  list
	 * @return void
	 */
	public void startListener(List<TopicTaskEntity> list){

		new Thread(new InterfaceListener(allURLService,applicationContext)).start();

		Iterator<TopicTaskEntity> iterator = list.iterator();
		while (iterator.hasNext()){
			TopicTaskEntity topicTaskEntity = iterator.next();
			if ("kafka".equals(topicTaskEntity.getTypeMq().toLowerCase())) {
				String broker = clusterService.selectById(topicTaskEntity.getClusterId()).getBroker();
				Properties consumerProps =kafkaConsumerService.generatorConsumerProps(broker,topicTaskEntity.getGroupId(),true);
				consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,100);
				KafkaConsumer<String, String>  kafkaConsumer = new KafkaConsumer<String, String>(consumerProps);
				// 订阅消息
				kafkaConsumer.subscribe(Collections.singletonList(topicTaskEntity.getTopicName()));

				//启动消息监听线程
				KafkaListenerJob kafkaListenerJob = new KafkaListenerJob(kafkaConsumerListener,kafkaConsumer,topicTaskEntity.getModelId(),topicTaskEntity.getModelName(),topicTaskEntity.getDataTable(),topicTaskEntity.getStructure(),topicTaskEntity.getQualityRule());
				Thread t = new Thread(kafkaListenerJob);
				t.start();

				topicTaskEntity.setThreadId(t.getId());

				topicTaskService.updateById(topicTaskEntity);
			} else if ("rabbitmq".equals(topicTaskEntity.getTypeMq().toLowerCase())) {
				RabbitMQEntity mqEntity = rabbitMQService.getById(topicTaskEntity.getClusterId());
				Connection connection = rabbitMQService.getConnection2(mqEntity);
				if (ObjectUtil.isNotEmpty(connection)){
					rabbitMQService.asyncGet(mqEntity,connection,topicTaskEntity);
				}
			}

		}
		/*Map<Object,List<TopicTaskEntity>> iterator = list.stream().collect(Collectors.groupingBy(TopicTaskEntity::getClusterId));
		Iterator<Map.Entry<Object,List<TopicTaskEntity>>> map= iterator.entrySet().iterator();
		while (map.hasNext()){

			Map.Entry<Object,List<TopicTaskEntity>> entry= map.next();
			long clusterId = Long.valueOf(entry.getKey().toString());
			List<TopicTaskEntity>  list1 = entry.getValue();
			String broker = clusterService.selectById(clusterId).getBroker();
			Properties consumerProps =kafkaConsumerService.generatorConsumerProps(broker,list1.get(0).getGroupId(),true);
			consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,100);
			KafkaConsumer<String, String>  kafkaConsumer = new KafkaConsumer<String, String>(consumerProps);
			List<String> stringList = list1.stream().map(TopicTaskEntity::getTopicName).collect(Collectors.toList());
			// 订阅消息
			kafkaConsumer.subscribe(stringList);
		}*/
	}


}
